package com.base.web.core.util.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * Created with IntelliJ IDEA.
 * Description: kafka生产者
 *
 * @Author: RenTie
 * @Date: 2018-08-20 11:35
 **/
@Service
public class KafkaProducer<K,V> {

    private Logger logger = LoggerFactory.getLogger(KafkaProducer.class);

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    /**
     * kafka发送消息
     * @param topic 主题
     * @param data  消息内容对象
     */
    public void send(String topic,V data) {

        String jsonObj = JSON.toJSONString(KafkaMessage.initMeassage(data));
        logger.info("------------ message = {}", jsonObj);

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, jsonObj);
        //发送回调
        callBack(future);
    }

    /**
     * kafka发送消息
     * @param topic 主题
     * @param key key
     * @param data 消息内容对象
     */
    public void send(String topic, String key, V data) {
        String jsonObj = JSON.toJSONString(KafkaMessage.initMeassage(data));
        logger.info("------------ message = {}", jsonObj);

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic,key,jsonObj);
        //发送回调
        callBack(future);
    }

    /**
     * kafka发送消息
     * @param topic 主题
     * @param partition 分区大小
     * @param key key
     * @param data 消息内容对象
     */
    public void send(String topic, Integer partition, String key, V data) {
        String jsonObj = JSON.toJSONString(KafkaMessage.initMeassage(data));
        logger.info("------------ message = {}", jsonObj);

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic,partition,key,jsonObj);
        //发送回调
        callBack(future);
    }

    /**
     * kafka发送消息
     * @param topic 主题
     * @param partition 分区大小
     * @param timestamp 时间戳
     * @param key key
     * @param data 消息内容对象
     */
    public void send(String topic, Integer partition, Long timestamp, String key, V data) {
        String jsonObj = JSON.toJSONString(KafkaMessage.initMeassage(data));
        logger.info("------------ message = {}", jsonObj);

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic,partition,timestamp,key,jsonObj);
        //发送回调
        callBack(future);
    }

    /**
     * kafka发送消息
     * @param record kafka消息实体
     */
    public void send(ProducerRecord<String, Object> record) {

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(record);
        //发送回调
        callBack(future);
    }

    public void send(Message<?> message) {
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(message);
        //发送回调
        callBack(future);
    }



    public void callBack(ListenableFuture<SendResult<String, Object>> future){
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            //失败回调
            @Override
            public void onFailure(Throwable throwable) {
                logger.info("Produce: The message failed to be sent:" + throwable.getMessage());
            }

            //成功回调
            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                //TODO 业务处理
                logger.info("Produce: The message was sent successfully");
                logger.info("Produce: _+_+_+_+_+_+_+ result: " + stringObjectSendResult.toString());
            }
        });
    }
}
